package com.atguigu.day11;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author Felix
 * @date 2024/4/13
 * 该案例演示了FlinkSQL表执行环境的准备
 *
 *  流--->动态表--->持续查询--->动态表--->Sink
 *                                 --->流
 */
public class FlinkSql01_Env {
    public static void main(String[] args) throws Exception {
        //TODO 1.~~~~~~~~~~~~~~~~基本环境准备~~~~~~~~~~~~~~~~
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(1);
        //1.3 指定表执行环境
        //StreamTableEnvironment.create(env, EnvironmentSettings.newInstance()
        //                .inStreamingMode()
        //        .build());
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2.~~~~~~~~~~~~~~~~创建表~~~~~~~~~~~~~~~~
        //~~~将流转换为表:方式1~~~
        //先准备一条流
        //DataStreamSource<Integer> numDS = env.fromElements(1, 2, 3, 4);
        //将流转换为动态表
        //Table test = tableEnv.fromDataStream(numDS,$("num"));
        //注意：直接通过这种方式查询是有问题的，Object 'test' not foun
        //tableEnv.executeSql("select * from test").print();
        //解决方式1  通过拼接对象的方式将对象注册到表执行环境中   底层调用Table对象的toString
        //tableEnv.executeSql("select * from " + test).print();
        //解决方式2 手动注册表对象到执行环境
        //tableEnv.createTemporaryView("test", test);
        //tableEnv.executeSql("select * from test").print();

        /*//~~~将流转换为表:方式2~~~
        //先准备一条流
        DataStreamSource<Integer> numDS = env.fromElements(1, 2, 3, 4);
        //将流转换为动态表
        tableEnv.createTemporaryView("t_test",numDS);
        tableEnv.executeSql("select * from t_test").print();

        //~~~将流转换为表:方式3 直接使用FlinkSQL提供的连接器~~~
        TableResult tableResult = tableEnv.executeSql("CREATE TABLE source ( \n" +
                "    id INT, \n" +
                "    ts BIGINT, \n" +
                "    vc INT \n" +
                ") WITH ( \n" +
                "    'connector' = 'datagen', \n" +
                "    'rows-per-second'='1', \n" +
                "    'fields.id.kind'='random', \n" +
                "    'fields.id.min'='1', \n" +
                "    'fields.id.max'='10', \n" +
                "    'fields.ts.kind'='sequence', \n" +
                "    'fields.ts.start'='1', \n" +
                "    'fields.ts.end'='1000000', \n" +
                "    'fields.vc.kind'='random', \n" +
                "    'fields.vc.min'='1', \n" +
                "    'fields.vc.max'='100'\n" +
                ")");
        tableEnv.executeSql("select * from source").print();*/

        //TODO 3.~~~~~~~~~~~~~~~~对表进行查询操作(持续查询)~~~~~~~~~~~~~~~~
        //DataStreamSource<Integer> numDS = env.fromElements(1, 2, 3, 4);
        //tableEnv.createTemporaryView("t_test",numDS,$("num"));
        //3.1 通过SQL方式进行查询---executeSql
        //TableResult tableResult = tableEnv.executeSql("select * from t_test");
        //tableResult.print();
        //3.2 通过SQL方式进行查询---sqlQuery
        //Table table = tableEnv.sqlQuery("select * from t_test where num=1");
        //tableEnv.createTemporaryView("test",table);
        //tableEnv.executeSql("select * from test").print();
        //table.execute().print();
        //3.2 通过API的方式进行查询   select * from t_test where num=1
        //Table test = tableEnv.sqlQuery("select * from t_test");
        //
        //test
        //        .select($("num"))
        //        .where($("num").isEqual(3))
        //        .execute()
        //        .print();


        //TODO 4.~~~~~~~~~~~~~~~~~~~~输出~~~~~~~~~~~~~~~~~~~~
        /*tableEnv.executeSql("CREATE TABLE source ( \n" +
                "    id INT, \n" +
                "    ts BIGINT, \n" +
                "    vc INT \n" +
                ") WITH ( \n" +
                "    'connector' = 'datagen', \n" +
                "    'rows-per-second'='1', \n" +
                "    'fields.id.kind'='random', \n" +
                "    'fields.id.min'='1', \n" +
                "    'fields.id.max'='10', \n" +
                "    'fields.ts.kind'='sequence', \n" +
                "    'fields.ts.start'='1', \n" +
                "    'fields.ts.end'='1000000', \n" +
                "    'fields.vc.kind'='random', \n" +
                "    'fields.vc.min'='1', \n" +
                "    'fields.vc.max'='100'\n" +
                ")");

        //TableResult tableResult = tableEnv.executeSql("select * from source where id =1");

        Table queryTable = tableEnv.sqlQuery("select * from source where id =1");
        //tableEnv.createTemporaryView("query_table",queryTable);


        tableEnv.executeSql("CREATE TABLE sink (\n" +
                "    id INT, \n" +
                "    ts BIGINT, \n" +
                "    vc INT\n" +
                ") WITH (\n" +
                "'connector' = 'print'\n" +
                ")\n");

        //tableEnv.executeSql("insert into sink select * from source");
        //tableEnv.executeSql("insert into sink select * from query_table");
        queryTable.executeInsert("sink");*/

        //TODO 5.~~~~~~~~~~~~~~~~~~~~将动态表转换为流~~~~~~~~~~~~~~~~~~~~
        /*tableEnv.executeSql("CREATE TABLE source ( \n" +
                "    id INT, \n" +
                "    ts BIGINT, \n" +
                "    vc INT \n" +
                ") WITH ( \n" +
                "    'connector' = 'datagen', \n" +
                "    'rows-per-second'='1', \n" +
                "    'fields.id.kind'='random', \n" +
                "    'fields.id.min'='1', \n" +
                "    'fields.id.max'='10', \n" +
                "    'fields.ts.kind'='sequence', \n" +
                "    'fields.ts.start'='1', \n" +
                "    'fields.ts.end'='1000000', \n" +
                "    'fields.vc.kind'='random', \n" +
                "    'fields.vc.min'='1', \n" +
                "    'fields.vc.max'='100'\n" +
                ")");
        //Table sourceTable = tableEnv.sqlQuery("select * from source");
        //toDataStream：追加操作
        //DataStream<Row> ds = tableEnv.toDataStream(sourceTable);
        //DataStream<WaterSensor> ds = tableEnv.toDataStream(sourceTable, WaterSensor.class);

        //Table queryTable = tableEnv.sqlQuery("select id,sum(vc) sumvc from source group by id");
        //toChangelogStream:可以有更新操作
        //DataStream<Row> ds = tableEnv.toChangelogStream(queryTable);
        //ds.print();*/


        //TODO 注意：在使用FlinkSQL的print方法的时候，方法会一直等到流中数据结束才会结束执行
        //+----+-------------+
        //| op |         num |
        //+----+-------------+
        //| +I |           1 |
        //| +I |           2 |
        //| +I |           3 |
        //| +I |           4 |
        //+----+-------------+
        //4 rows in set
        //DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 8888);
        //tableEnv.createTemporaryView("socket_table",socketDS,$("num"));
        //tableEnv.executeSql("select * from socket_table").print();
        //tableEnv.executeSql("select * from socket_table where num='100'").print();


        //TODO 注意：如果操作的流，需要通过env.execute提交作业，如果程序最后操作的是动态表，不要调用该方法
        env.execute();


    }
}
